// package com.dt.spark.SparkApps.sql;
// /**
//  * 打印结果：
//         [1,Spark,7]
//         [2,Hadoop,10]
//         [3,Flink,5]
//  */
import java.util.ArrayList;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
// import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

public class RDDToDataFrameByProgrammatically 
{

    public static void main(String[] args) 
    {
        SparkConf conf = new SparkConf().setMaster("local").
        setAppName("RDDToDataFrameByReflection");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        //读取数据
        JavaRDD<String> lines = sc.textFile("persons.txt");
        /**
         * 第一步：在RDD的基础上创建类型为Row的RDD
         */
        //首先，必须将RDD变成以Row为类型的RDD。Row可以简单理解为Table的一行数据
        JavaRDD<Row> personsRDD = lines.map(new Function<String,Row>()
        {

            @Override
            public Row call(String line) throws Exception 
            {
                String[] splited = line.split(",");
                return RowFactory.create(Integer.valueOf(splited[0]),splited[1],Integer.valueOf(splited[2]));
            }

        });

        /**
         * 第二步：动态构造DataFrame的元数据，一般而言，有多少列以及每列的具体类型可能来自于
         * JSON文件，也可能来自于DB
         */
        //对Row具体指定元数据信息。
        List<StructField> structFields = new ArrayList<StructField>();
        //列名称  列的具体类型（Integer Or String） 是否为空一般为true，实际在开发环境是通过for循环，而不是手动添加
        structFields.add(DataTypes.createStructField("id", DataTypes.IntegerType, true));
        structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
        structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
        //构建StructType,用于最后DataFrame元数据的描述
        StructType structType = DataTypes.createStructType(structFields);

//-----------------------------------------------------------------------------------------------------------------

            // 第三步：基于已有的MetaData以及RDD<Row>来构造DataFrame
            Dataset<Row> df=sqlContext.createDataFrame(personsRDD, structType);//这句话就是使用反射的语句,注意第二个参数Person.class
    //这个其实就是利用反射来提供字段信息,专业点的说法,就是利用反射(Person.class)来提供元数据(字段)


        // 第四步：注册成临时表以供后续的SQL查询操作
        df.registerTempTable("persons"); 


        // 第五步：进行数据的多维度分析
        Dataset<Row> bigdatas=sqlContext.sql("select * from persons where age>=6");

        System.out.println("----------------------------------下面的是临时注册表输出结果---------------------------------------");
        bigdatas.show();

        System.out.println("----------------------------------下面的是Dataframe转化成RDD并输出---------------------------------------");
            // 第六步：对结果进行处理，包括由DataFrame转换成为RDD<Row>,以及结果的持久化

        List<Row> listRow = bigdatas.javaRDD().collect();
        for(Row row : listRow)
        {
            System.out.println(row);
        }

    }

}